import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.net.DatagramSocket;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.DatagramChannel;
import java.nio.channels.SelectableChannel;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.nio.channels.spi.SelectorProvider;
import java.util.*;

import org.mybeans.dao.DAOException;
import org.mybeans.factory.BeanTable;

import serverMessage.AskUserPubKey;
import serverMessage.FindNeighborMessage;
import serverMessage.RegisterMERequest;
import serverMessage.RegisterMEResponse;
import serverMessage.ServerMessage;
import serverMessage.UserCreateSessionInfoRequestRecord;
import serverMessage.UserLoginAnnouncement;
import serverMessage.UserPubKey;
import serverMessage.UserRegistrationRecord;
import serverMessage.WriteCommand;
import time.LogicalClock;
import time.LogicalTimeStamp;
import time.VectorClock;
import time.VectorTimeStamp;

import databeans.DroidChatUserBean;

import edu.cmu.ece.ds.droidchat.message.*;

import model.DroidChatUserDAO;

public class NioServer implements Runnable {
	private int port1; // port to listen to other DroidChat servers
	private int port2; // port to listen to DroidChat clients
	
	public static final int udp_port = 5432; 		// my udp port number
	public static final int client_udpport = 6666;	// client's udp port number
	
	private String myAddress;
	
	private ServerSocketChannel serverChannel1; // listen to server connections
	private ServerSocketChannel serverChannel2; // listen to client connections
	
	private Selector selector;
	
	// information about DroidChat servers
	private HashMap<String, Server> serverInfo;
	private Stack<String> deadServer;
	
	private Properties properties;
	private String myServerName; // my DroidChat server name
	private int myPos; // my process id
	private ClientMessageWorker worker;
	
	private DroidChatUserDAO userDAO = null;
	private LinkedList<ChangeRequest> changeRequests;
	
	// maps a Channel to a list of DroidChatMessages to be sent
	private HashMap<SelectableChannel, LinkedList<ByteBuffer>> pendingData;
	
	// maps a SocketChannel to the current incoming message to be read
	private HashMap<SocketChannel, LinkedList<ByteBuffer>> pendingIncomingData;
	// [0] is the byteSize header  - should be 4 bytes
	// [1] is the actually message - should be as long as speciied in 0th element
	
	// maps a login_session_id to login information to keep at the server side for each logged in user
	private HashMap<Integer,Login_Session_Attr> userLoginRecord;
	
	// a list of ClientRegisterMessage received
	private ArrayList<UserRegistrationRecord> registrationRequests;
	
	private VectorClock MEClock; // vector clock used to ensure MERequests are causally ordered
	private ArrayList<WorkerData> holdbackQueue; // if RegisterMERequest cannot yet be delivered
												 //  they will go into this holdback queue
												 //  the WorkData in the holdbackQueue contains message Type RegisterMERequest
	private LogicalClock logicalClock;
	
	private String writeMEState = RELEASED; // write mutual exclusion state
	
	public static final String HELD = "HELD";
	public static final String WANTED = "WANTED";
	public static final String RELEASED = "RELEASED";
	
	private LogicalTimeStamp lastMEReq = null;
	
	private DatagramChannel UDPchannel;
	private DatagramSocket UDPsocket;
	
	public static final double fake_latitude = 40.44208;
	public static final double fake_longitude = -79.9469679;
	
	public static final double neighbor_distance = 0.5;
	
	public static enum Availability {
        AVAILABLE, BUSY
    }
	
	/**
	 * Inner class ClientMessageWorker
	 * 
	 */
	private class ClientMessageWorker implements Runnable{
		private LinkedList<WorkerData> queue = new LinkedList<WorkerData>();// a queue of work to be processed by this thread
		private DroidChatUserDAO userDAO;// the data access object to interact with the database
		private Random r;// used to generate the randomly generated login_session_id
		
		private LinkedList<UserCreateSessionInfoRequestRecord> chatSessionInfoRequest; 
		
		private long lastStaleUserCheckTime=0;
		
		public ClientMessageWorker(DroidChatUserDAO userDAO) {
			 this.userDAO = userDAO;
			 r = new Random();
			 chatSessionInfoRequest = new LinkedList<UserCreateSessionInfoRequestRecord>();
		}

		// the selecting thread calls this method
		//  to give the worker 'work' to do
		public void processReceivedDroidChatMessage(NioServer server, SocketChannel socketChannel, DroidChatMessage message) {
			synchronized(queue) {
				queue.add(new WorkerData(server, socketChannel, message));
				queue.notify();
			}
		}/* end of processReceivedDroidChatMessage() */
		
		// the selecting thread calls this method
		public void processReceivedServerMessage(NioServer nioServer, SocketChannel socketChannel, ServerMessage message, Server messageSource){
			synchronized(queue) {
				WorkerData work = new WorkerData(nioServer, socketChannel, message);
				work.setMessageSource(messageSource.getServerName()); // worker thread now knows which server sent this ServerMessage
				queue.add(work);
				queue.notify();
			}
		}/* end of processReceivedServerMessage() */
		
		// the selecting thread calls this method
		public void insertRegisterMERequest(NioServer nioServer, SocketChannel socketChannel, RegisterMERequest MEreq, Server messageSource) {
			VectorTimeStamp msgTS = MEreq.getME_TS();
			boolean inserted = false;
			
			WorkerData newWork = new WorkerData(nioServer, socketChannel, MEreq);
			newWork.setMessageSource(messageSource.getServerName());
			
			synchronized(holdbackQueue) {
				WorkerData work = new WorkerData(nioServer, socketChannel, MEreq);
				work.setMessageSource(messageSource.getServerName()); // worker thread now knows which server sent this ServerMessage
				
				// insert in order
				for(int i=0; i<holdbackQueue.size(); i++){
					RegisterMERequest cur = (RegisterMERequest) holdbackQueue.get(i).getMessage();
					
					if(msgTS.happenedBefore(cur.getME_TS())){
						holdbackQueue.add(i, newWork);
						inserted=true;
						break;
					}
				}
				if(inserted==false){
					holdbackQueue.add(newWork);
					inserted=true;
				}
			}
		}
		
		// this is the execution of the worker thread, 
		//  NioServer.sendToClient() is called here by the worker thread which wakes up the 
		//  the selecting thread
		public void run() {
			
			WorkerData dataEvent = null; // the current work to process
			
			while(true) {
				synchronized (holdbackQueue) {
					if(!holdbackQueue.isEmpty()){
						if(writeMEState.equals(RELEASED)){
							RegisterMERequest req = (RegisterMERequest) holdbackQueue.get(0).getMessage();
							if(canProcess(req)){
								System.out.println(holdbackQueue.get(0).getMessage() + " can be processed now");
								synchronized(queue){
									queue.add(holdbackQueue.remove(0));
								}
							}
						}
					} else {
						// was there any new RegisterRequest that I can now send out
						synchronized (registrationRequests) {
							if(!registrationRequests.isEmpty() && registrationRequests.get(0).getRequestTime()==null && writeMEState.equals(RELEASED)){
								// registration request that could not be multicasted is now being processed
								UserRegistrationRecord r = registrationRequests.get(0);
								
								System.out.println("	registration request that could not be multicasted is now being processed");
								System.out.println("before="+registrationRequests.get(0));
								MEClock.timestampNextEvent();
								logicalClock.timestampNextEvent();
								RegisterMERequest registerReq = new RegisterMERequest(MEClock.getTimeStamp(), r.getUsername(), logicalClock.getTimeStamp());
								lastMEReq = logicalClock.getTimeStamp();
								multicastToServers(registerReq);
								writeMEState = WANTED;
								System.out.println("	ME state=WANTED");
								registrationRequests.get(0).setRequestTime(lastMEReq);
								System.out.println("after="+registrationRequests.get(0));
							} else if (!registrationRequests.isEmpty() && registrationRequests.get(0).getRequestTime()!=null && writeMEState.equals(WANTED)) {
								System.out.println("	I'm still waiting for some vote");
								// i am still waiting for some vote
								UserRegistrationRecord r = registrationRequests.get(0);
								
								if(allLiveServers(r.getAgreedServers())){
									// obtained the write ME
									writeMEState = HELD;
									System.out.println("allLiveServers:" + r.getAgreedServers());
									System.out.println("write mutual exclusion obtained for " + r);
									
									// multicast write command to all servers
									logicalClock.timestampNextEvent();
									WriteCommand command = new WriteCommand(r.getUsername(), r.getPassword(), logicalClock.getTimeStamp());
									command.setBytes(r.getBytes());
									command.setBG(r.getBG());
									command.setNickname(r.getNickname());
									multicastToServers(command);
									
									// write into local database
									DroidChatUserBean newUser = new DroidChatUserBean();
									newUser.setUsername(r.getUsername());
									newUser.setPassword(r.getPassword());
									newUser.setBytes(r.getBytes());
									newUser.setAvailability(Availability.AVAILABLE.ordinal());
									newUser.setBG(r.getBG());
									newUser.setNickname(r.getNickname());
									
									DroidChatUserBean createdUser = null;
									
									try {
										createdUser = userDAO.create(newUser);
									} catch (DAOException e) {	e.printStackTrace(); }
									
									System.out.println("written into database:" + createdUser);
									
									ClientRegisterResponseMessage responseToClient = 
										new ClientRegisterResponseMessage(ClientRegisterResponseMessage.REGISTER_SUCCESSFUL,
												createdUser.getUsername());
										
									// reply to client that command
									sendToClient(r.getSocket(), responseToClient);
									registrationRequests.remove(0); // done with this registration request
									
									// now that I'm released 
									writeMEState = RELEASED;
									
									// was there any pending MErequests for me to process?
									System.out.println("holdback queue size: " + holdbackQueue.size());
										
									if(!holdbackQueue.isEmpty()){
										RegisterMERequest req = (RegisterMERequest) holdbackQueue.get(0).getMessage();
										if(canProcess(req)){
											System.out.println(holdbackQueue.get(0).getMessage() + " can be processed now");
											synchronized(queue){
												queue.add(holdbackQueue.remove(0));
											}
										}
									} else {
										// was there any new RegisterRequest that I can now send out
										synchronized (registrationRequests) {
											if(!registrationRequests.isEmpty() && registrationRequests.get(0).getRequestTime()==null){
												// registration request that could not be multicasted is now being processed
												System.out.println("	registration request that could not be multicasted is now being processed");
												System.out.println("	before="+registrationRequests.get(0));
												MEClock.timestampNextEvent();
												logicalClock.timestampNextEvent();
												RegisterMERequest registerReq = new RegisterMERequest(MEClock.getTimeStamp(), r.getUsername(), logicalClock.getTimeStamp());
												lastMEReq = logicalClock.getTimeStamp();
												multicastToServers(registerReq);
												writeMEState = WANTED;
												System.out.println("	ME state=WANTED");
												registrationRequests.get(0).setRequestTime(lastMEReq);
												System.out.println("	after="+registrationRequests.get(0));
											}
										}
									}
								}
							}
						}
					}
				}
				
				// Wait for data to become available
				//  (i.e., sleep until there is work for me to process)
				synchronized(queue) {
					while(queue.isEmpty()) {
						try {
							queue.wait();
						}catch (InterruptedException e) {}
					}
					dataEvent = queue.remove();
				}
				
				if(dataEvent.getMessage() instanceof DroidChatMessage){
					/** message from Client **/
					if (dataEvent.getMessage() instanceof ClientLoginMessage) {
						processClientLoginMessage(dataEvent);
					} else if (dataEvent.getMessage() instanceof ClientRegisterMessage) {
						processClientRegisterMessage(dataEvent);
					} else if(dataEvent.getMessage() instanceof ClientPublicKeyMessage) {
						processClientPublicKeyMessage(dataEvent);
					} else if(dataEvent.getMessage() instanceof ClientUpdateLocationMessage) {
						evictStaleLoggedInUsers();
						processClientUpdateLocationMessage(dataEvent);
					} else if(dataEvent.getMessage() instanceof CreateSessionInfoRequestMessage){
						processCreateSessionInfoRequestMessage(dataEvent);
					/*} else if(dataEvent.getMessage() instanceof CreateSessionInfoRequestMessageR){
						processCreateSessionInfoRequestMessageR(dataEvent);*/
					}else if(dataEvent.getMessage() instanceof LogOutRequestFromClient){
						processLogOutRequestFromClient(dataEvent);
					}
				} else if(dataEvent.getMessage() instanceof ServerMessage) {
					/** message from Server **/
					if (dataEvent.getMessage() instanceof RegisterMERequest) {
						processRegisterMERequest(dataEvent);
					} else if (dataEvent.getMessage() instanceof RegisterMEResponse){
						processRegisterMEResponse(dataEvent);
					} else if (dataEvent.getMessage() instanceof WriteCommand) {
						processWriteCommand(dataEvent);
					} else if (dataEvent.getMessage() instanceof FindNeighborMessage){
						evictStaleLoggedInUsers();
						processFindNeighborMessage(dataEvent);
					} else if (dataEvent.getMessage() instanceof AskUserPubKey) {
						processAskUserPubKey(dataEvent);
					} else if (dataEvent.getMessage() instanceof UserPubKey){
						processUserPubKey(dataEvent);
					} else if (dataEvent.getMessage() instanceof UserLoginAnnouncement){
						processUserLoginAnnouncement(dataEvent);
					}
				}
			}
			
		}/* end of run() */


		/**
		 * worker thread helper method: processLogOutRequestFromClient()
		 */
		private void processLogOutRequestFromClient(WorkerData dataEvent) {
			System.out.println("----processLogOutRequestFromClient()----");
			
			LogOutRequestFromClient m = (LogOutRequestFromClient) dataEvent.getMessage();
			
			int loginSessionID = m.getLogin_session_id();
			
			if(userLoginRecord.containsKey(loginSessionID)){
				userLoginRecord.remove(loginSessionID);
				System.out.println("successfully removed the loginSessionID " + loginSessionID);
			}
		}

		/**
		 * worker thread helper method: evictStaleLoggedInUsers()
		 * 	called right before processFindNeighborMessage() or 
		 *  processClientUpdateLocationMessage() where location of
		 *  some user may need to be returned
		 *  
		 * it looks at lastUpdateTime field of each Login_Session_Attr for each
		 *  logged in user, if too old, remove it from hashtable 'userLoginRecord'
		 */
		private void evictStaleLoggedInUsers() {
			long timeDiff = System.currentTimeMillis() - lastStaleUserCheckTime;
			
			if(timeDiff > Login_Session_Attr.LOCATION_TIMEOUT){
				Iterator<Login_Session_Attr> itr = userLoginRecord.values().iterator();
				
				//System.out.println("		b4 key size" + userLoginRecord.size());
				while(itr.hasNext()){
					Login_Session_Attr cur = itr.next();
					
					if(cur.getRound() > 0){
						timeDiff = System.currentTimeMillis() - cur.getLastUpdateTime();
						
						if(timeDiff > Login_Session_Attr.LOCATION_TIMEOUT){
							itr.remove();
						}
					}
				}
				//System.out.println("		after key size" + userLoginRecord.size());
			}
		}
		
		/**
		 * worker thread helper method: processUserPubKey()
		 */
		private void processUserPubKey(WorkerData dataEvent) {
			UserPubKey m = (UserPubKey) dataEvent.getMessage();
			
			Iterator<UserCreateSessionInfoRequestRecord> itr = chatSessionInfoRequest.iterator();
			
			UserCreateSessionInfoRequestRecord req = null; itr.next();
			
			boolean found = false;
			
			while(itr.hasNext()){
				req = itr.next();
				if(req.getUsername().equals(m.getUsername())){
					found = true;
					break;
				}
			}
			
			if(found){
				// TODO: ASKPUBKEY needs to be changed as well
				//CreateSessionInfoResponseMessage response = new CreateSessionInfoResponseMessage(chatSessionid, modulus, exponent, username)
				/*CreateSessionInfoResponseMessage response = new CreateSessionInfoResponseMessage(m.getPublic_modulus(),
						m.getPublic_exponent(), m.getUsername());
				sendToClient(req.getSocket(), response);
				itr.remove(); // remove from recordI*/
			}
		}

		/**
		 * worker thread helper method: processAskUserPubKey()
		 */
		private void processAskUserPubKey(WorkerData dataEvent) {
			AskUserPubKey m = (AskUserPubKey) dataEvent.getMessage();
			String username = m.getUsername();
			
			// look inside userLoginRecord and see if username is one of my logged-in client
			//  if so reply with UserPubKey
			for(Login_Session_Attr user: userLoginRecord.values()){
				if(user.getUsername().equals(username)){
					UserPubKey response = new UserPubKey(username,
							user.getModulus(),
							user.getExponent(),
							true,
							logicalClock.timestampNextEvent());

					Server s = null;
					String source = dataEvent.getServerName();
					
					synchronized (serverInfo) {
						if(serverInfo.containsKey(source)){
							s = serverInfo.get(source);
						}
					}
					
					if(s==null)
						return;
					
					sendToServer(s.getChannel(), response);
				}
			}
			return;
		}

		/*private void processCreateSessionInfoRequestMessageR(WorkerData dataEvent) {
			System.out.println("processCreateSessionInfoRequestMessageR() ");
			
			CreateSessionInfoRequestMessageR m = (CreateSessionInfoRequestMessageR) dataEvent.getMessage();
			String username = m.getUsername();
			
			// look inside userLoginRecord and see if username is one of my logged-in client
			//  if so reply with CreateSessionInfoResponse
			for(Login_Session_Attr user: userLoginRecord.values()){
				System.out.println("user " + user.getUsername());
				
				if(user.getUsername().equals(username)){
					//CreateSessionInfoResponseMessage response = new CreateSessionInfoResponseMessage(user.getModulus(), user.getExponent(), user.getUsername());
					CreateSessionInfoResponseMessageR response = new CreateSessionInfoResponseMessageR(
							m.getChatSessionID(),
							user.getModulus(),
							user.getExponent(),
							user.getUsername());
					sendToClient(dataEvent.getSocket(), response);
					System.out.println("\n found " + user.getUsername()
							+ "modulus" + user.getModulus());
					return;
				}
			}
			
			// else seek help with other servers
			
			
			// first record it 
			UserCreateSessionInfoRequestRecord record = new UserCreateSessionInfoRequestRecord(
					dataEvent.getSocket(),
					username);
			chatSessionInfoRequest.add(record);
			
			AskUserPubKey ask = new AskUserPubKey(username, logicalClock.timestampNextEvent());
			multicastToServers(ask);
		}*/
		
		/**
		 * worker thread helper method: processCreateSessionInfoRequestMessage()
		 */
		private void processCreateSessionInfoRequestMessage(WorkerData dataEvent) {
			System.out.println("processCreateSessionInfoRequestMessage() ");
			
			CreateSessionInfoRequestMessage m = (CreateSessionInfoRequestMessage) dataEvent.getMessage();
			String username = m.getUsername();
			
			System.out.println("asking for " + username);
			
			// look inside userLoginRecord and see if username is one of my logged-in client
			//  if so reply with CreateSessionInfoResponse
			for(Login_Session_Attr user: userLoginRecord.values()){
				System.out.println("user " + user.getUsername());
				
				if(user.getUsername().equals(username)){
					//CreateSessionInfoResponseMessage response = new CreateSessionInfoResponseMessage(user.getModulus(), user.getExponent(), user.getUsername());
					CreateSessionInfoResponseMessage response = new CreateSessionInfoResponseMessage(
							m.getChatSessionID(),
							user.getModulus(),
							user.getExponent(),
							user.getUsername());
					sendToClient(dataEvent.getSocket(), response);
					System.out.println("\n found " + user.getUsername()
							+ "modulus" + user.getModulus());
					return;
				}
			}
			
			// else seek help with other servers
			
			
			// first record it 
			UserCreateSessionInfoRequestRecord record = new UserCreateSessionInfoRequestRecord(
					dataEvent.getSocket(),
					username);
			chatSessionInfoRequest.add(record);
			
			AskUserPubKey ask = new AskUserPubKey(username, logicalClock.timestampNextEvent());
			multicastToServers(ask);
		}

		/**
		 * worker thread helper method: processFindNeighborMessage()
		 */
		private void processFindNeighborMessage(WorkerData dataEvent) {
			FindNeighborMessage findNeighbor = (FindNeighborMessage) dataEvent.getMessage();
			
			double requesterLatitude = findNeighbor.getLatitude();
			double requesterLongitude = findNeighbor.getLongitude();
			int requesterRound = findNeighbor.getRoundNumber();

			// I am a server, I looking at my list of logged-in users to help the requestor
			//  build its neighbor list
			for(int session_id: userLoginRecord.keySet()){
				Login_Session_Attr user = userLoginRecord.get(session_id);
				if(user.getRound() > 0){
					
					double distance = distance(requesterLatitude, requesterLongitude, user.getLast_known_lattitude(), user.getLast_known_longitude());
					System.out.println(" distance is " + distance);
					
					if(distance < neighbor_distance ){
						System.out.println("	a nearby user is found. " + user.getUsername());
						
						// a neighbor is found, send out an UDP NeighborInfo back to currentUser
						NeighborInfoMessage neighbor = new NeighborInfoMessage(user.getUsername(), user.getIP());
						neighbor.setRound(requesterRound);
						
						DroidChatUserBean u = null;
						try {
							u = userDAO.lookup(user.getUsername());
						} catch (DAOException e) {
							e.printStackTrace();
						}
						
						if(u!=null){
							System.out.println("picture byte size=" + u.getBytes().length);
							neighbor.setBytes(u.getBytes());
							neighbor.setBG(u.getBG());
							neighbor.setNickname(u.getNickname());
						}
						
						sendToClientUDP(neighbor, findNeighbor.getSrcIP());			
					}
				}
			}
		}

		/**
		 * worker thread helper method: processClientUpdateLocationMessage()
		 */
		private void processClientUpdateLocationMessage(WorkerData dataEvent) {
			ClientUpdateLocationMessage update = (ClientUpdateLocationMessage) dataEvent.getMessage();
			int login_session_id = update.getLogin_session_id();
			
			Login_Session_Attr currentUser = null;
			
			if(userLoginRecord.containsKey(login_session_id)){
				currentUser = userLoginRecord.get(login_session_id);
				currentUser.setLast_known_lattitude(update.getLatitude());
				currentUser.setLast_known_longitude(update.getLongitude());
				currentUser.setRound(update.getRound_number());
				currentUser.setLastUpdateTime(System.currentTimeMillis());
			}else{
				// reply with an LoggedOutMessage
				//  so that the phone UI can react accordingly
				LogOutCommandFromServer logout = new LogOutCommandFromServer(login_session_id);
				sendToClient(dataEvent.getSocket(), logout);
				return;
			}
			
			// multicast to all servers that this user needs to find its neighbors
			logicalClock.timestampNextEvent();
			FindNeighborMessage findNeighbor = new FindNeighborMessage(logicalClock.getTimeStamp(),
					currentUser.getIP(),
					currentUser.getLast_known_longitude(),
					currentUser.getLast_known_lattitude(), 
					currentUser.getRound());
			
			dataEvent.getServer().multicastToServers(findNeighbor);
			
			System.out.println("total number of users:" + userLoginRecord.keySet());
			System.out.println("for user " + currentUser.getUsername());
			
			for(int session_id: userLoginRecord.keySet()){
				Login_Session_Attr user = userLoginRecord.get(session_id);
				
				if(currentUser!=user && user.getRound() > 0){
					double d = distance(currentUser.getLast_known_lattitude(), currentUser.getLast_known_longitude(), user.getLast_known_lattitude(), user.getLast_known_longitude());
					
					System.out.println("distance is " + d);
					
					if(d < neighbor_distance ){
						System.out.println("	a nearby user is found. " + user.getUsername());
						
						// a neighbor is found, send out an UDP NeighborInfo back to currentUser
						NeighborInfoMessage neighbor = new NeighborInfoMessage(user.getUsername(), user.getIP());
						neighbor.setRound(currentUser.getRound());
						
						DroidChatUserBean u = null;
						try {
							u = userDAO.lookup(user.getUsername());
						} catch (DAOException e) {
							e.printStackTrace();
						}
						
						if(u!=null){
							System.out.println("picture byte size=" + u.getBytes().length);
							neighbor.setBytes(u.getBytes());
							neighbor.setBG(u.getBG());
							neighbor.setNickname(u.getNickname());
						}
						
						//System.out.println("	" + neighbor);
						System.out.println("	sendNeighborInfo() to " + currentUser.getIP());
						sendToClientUDP(neighbor, currentUser.getIP());					
					}
				}
			}
		}

		/**
		 * worker thread helper method: processClientPublicKeyMessage()
		 */
		private void processClientPublicKeyMessage(WorkerData dataEvent) {
			System.out.println("processing a ClientPublicKeyMessage");
			synchronized(userLoginRecord){
				ClientPublicKeyMessage message = (ClientPublicKeyMessage)dataEvent.getMessage();
				if(userLoginRecord.containsKey(message.getLogin_session_id())){
					Login_Session_Attr attr = userLoginRecord.get(message.getLogin_session_id());
					attr.setExponent(message.getExponent());
					attr.setModulus(message.getModulus());
					
					InetAddress clientAddress = dataEvent.getSocket().socket().getInetAddress();
					System.out.println("	" + clientAddress.getHostAddress());
					attr.setIP(clientAddress.getHostAddress());
				}
			}
		}

		/**
		 * worker thread helper method: processUserLoginAnnouncement()
		 *  
		 * it looks at the logged in user record and if the username is found, that
		 *  means the user has logged in again at another server
		 */
		private void processUserLoginAnnouncement(WorkerData dataEvent) {
			UserLoginAnnouncement announcment = (UserLoginAnnouncement) dataEvent.getMessage();
			
			Iterator<Login_Session_Attr> itr = userLoginRecord.values().iterator();
			
			while(itr.hasNext()){
				Login_Session_Attr login = itr.next();
				
				if(login.getUsername().equals(announcment.getUsername())){
					// log this guy out first
					LogOutCommandFromServer logout = new LogOutCommandFromServer(login.getLogin_session_id());
					System.out.println(" sending out LogOutCommandFromServer ");
					// by removing the old user from userLoginRecord, when the next location update comes
					//  the login_session_id included in the message won't be recognized,
					//  and he will end up receiving a LogOutCommandFromServer
					itr.remove();
				}
			}
			
		}
		
		/**
		 * worker thread helper method: processClientLoginMessage()
		 */
		private void processClientLoginMessage(WorkerData work) {
			// processing a ClientLoginMessage
			ClientLoginMessage m = (ClientLoginMessage) work.getMessage();
			DroidChatUserBean u = null;
			
			try {
				u = userDAO.lookup(m.getUsername());
			} catch (DAOException e) { e.printStackTrace(); }
			
			if (u==null) {
				// no user with such username is found
				System.out.println("	worker thread: no such username=" + m.getUsername());
				ClientLoginResponseMessage response = new ClientLoginResponseMessage(m.getUsername(), ClientLoginResponseMessage.NO_USER);
				work.getServer().sendToClient(work.getSocket(), response);
			} else {
				if (u.getPassword().equals(m.getPassword())) {
					// successful login
					
					Iterator<Login_Session_Attr> itr = userLoginRecord.values().iterator();
					
					while(itr.hasNext()){
						Login_Session_Attr login = itr.next();
						
						if(login.getUsername().equals(u.getUsername())){
							// log this guy out first
							LogOutCommandFromServer logout = new LogOutCommandFromServer(login.getLogin_session_id());
							System.out.println(" sending out LogOutCommandFromServer ");
							sendToClient(work.getSocket(), logout);
							itr.remove();
						}
					}
					
					ClientLoginResponseMessage response = new ClientLoginResponseMessage(m.getUsername(), ClientLoginResponseMessage.LOGIN_SUCCESSFUL);
					response.setBytes(u.getBytes());
					response.setBG(u.getBG());
					response.setNickname(u.getNickname());
					
					int loginid=r.nextInt();
					response.setLoginSessionID(loginid); // a randomly generated login_session_id
					work.getServer().sendToClient(work.getSocket(), response);
					userLoginRecord.put(loginid, new Login_Session_Attr(loginid,u.getUsername()));
					
					// send to all servers that this user just logged in -> 
					//  to prevent multiple logins of the same user
					logicalClock.timestampNextEvent();
					UserLoginAnnouncement anountment = new UserLoginAnnouncement(m.getUsername(), logicalClock.getTimeStamp());
					work.getServer().multicastToServers(anountment);
				} else {
					// login fails b/c of wrong password
					ClientLoginResponseMessage response = new ClientLoginResponseMessage(m.getUsername(), ClientLoginResponseMessage.WRONG_PASSWORD);
					work.getServer().sendToClient(work.getSocket(), response);
				}
			}
		}
		
		/**
		 * worker thread helper method: processClientRegisterMessage()
		 */
		private void processClientRegisterMessage(WorkerData work) {
			System.out.println("******* processClientRegisterMessage *******");
			ClientRegisterMessage m = (ClientRegisterMessage) work.getMessage();
			DroidChatUserBean u = null;
			
			try {
				u = userDAO.lookup(m.getUsername());
			} catch (DAOException e) { e.printStackTrace(); }
			
			if (u==null) {
				// no one has tried registering with this username
				System.out.println("	worker thread (RegisterMessage): username=" + m.getUsername() + " password=" + m.getPassword() + " not registered yet");
				System.out.println("    but still has to request ME to write ");
				
				UserRegistrationRecord record = new UserRegistrationRecord(work.getSocket(), m.getUsername(), m.getPassword());
				
				synchronized (writeMEState) {	
					if (writeMEState.equals(RELEASED)) {
						System.out.println("	ME state=RELEASED");
						
						MEClock.timestampNextEvent();
						logicalClock.timestampNextEvent();
						RegisterMERequest registerReq = new RegisterMERequest(MEClock.getTimeStamp(), m.getUsername(), logicalClock.getTimeStamp());
						
						lastMEReq = logicalClock.getTimeStamp();
						work.getServer().multicastToServers(registerReq);
						writeMEState = WANTED;
						System.out.println("	ME state=WANTED");
						
						record.setRequestTime(lastMEReq);
						record.addAgreedServer(myServerName);
						// new Set the picture
						record.setBytes(m.getPic());
						record.setBG(m.getBG());
						record.setNickname(m.getNickname());
						
						System.out.println("\n\n\n\n"+ m.getBG());
						
						System.out.println("adding into registrationRequests [" + lastMEReq + "]: " + record);
						registrationRequests.add(record);
						
					} else {
						System.out.println("	ME state=" + writeMEState);
						// either WANTED or HELD state, cannot multicast request now
						//  queue up the registration request
						record.addAgreedServer(myServerName);
						// record.requestTime would be null because no request made yet
						System.out.println("adding into registrationRequests [null]: " + record);
						//System.exit(0);
						registrationRequests.add(record);
					} 
				}
				
			} else {
				// username has already been registered -> registration fails
				ClientRegisterResponseMessage response = new ClientRegisterResponseMessage(ClientRegisterResponseMessage.USERNAME_ALREADY_REGISTERED, m.getUsername());
				work.getServer().sendToClient(work.getSocket(), response);
			}
		}

		/**
		 * worker thread helper method: processRegisterMERequest()
		 */
		private void processRegisterMERequest(WorkerData work) {
			
			// RegisterMERequest
			RegisterMERequest registerReq = (RegisterMERequest) work.getMessage();
			
			// I am processing a RegisterMERequest message, have to update MEClock
			synchronized (MEClock) {
				MEClock.updateUponDeliver(registerReq.getME_TS());
			}
			System.out.println("\nprocessing " + registerReq + " from " + work.getServerName());
			
			DroidChatUserBean user = null;
			
			try {
				user = userDAO.lookup(registerReq.getUsername());
			} catch (DAOException e) {
				e.printStackTrace();
			}
				
			if(writeMEState.equals(RELEASED)){
				System.out.println(writeMEState);
				if (user==null) {
					// user not yet registered in my local database
					System.out.println("user not yet registered in my local database");
					logicalClock.timestampNextEvent();
					RegisterMEResponse registerResponse = new RegisterMEResponse(true, registerReq.getRequestTime(), logicalClock.getTimeStamp());
					System.out.println(registerResponse);
					work.getServer().sendToServer(work.getSocket(), registerResponse);
				} else { 
					// user already registered in my local database
					System.out.println("user already registered in my local database");
					logicalClock.timestampNextEvent();
					RegisterMEResponse registerResponse = new RegisterMEResponse(false, registerReq.getRequestTime(), logicalClock.getTimeStamp());
					System.out.println(registerResponse);
					work.getServer().sendToServer(work.getSocket(), registerResponse);
				}
			} else {
				System.out.println("I must be in WANTED State and has lastReq that's larger");
				System.out.println(writeMEState);
				System.out.println("lastMEReq=" + lastMEReq);
				System.out.println("incomingMEreq=" + registerReq.getME_TS());
				
				if (user==null) {
					// user not yet registered in my local database
					System.out.println("user not yet registered in my local database");
					logicalClock.timestampNextEvent();
					RegisterMEResponse registerResponse = new RegisterMEResponse(true, registerReq.getRequestTime(), logicalClock.getTimeStamp());
					System.out.println(registerResponse);
					work.getServer().sendToServer(work.getSocket(), registerResponse);
				} else {
					// user already registered in my local database
					System.out.println("user already registered in my local database");
					logicalClock.timestampNextEvent();
					RegisterMEResponse registerResponse = new RegisterMEResponse(false, registerReq.getRequestTime(), logicalClock.getTimeStamp());
					System.out.println(registerResponse);
					work.getServer().sendToServer(work.getSocket(), registerResponse);
				}
			}
		}

		/**
		 * worker thread helper method: processRegisterMEResponse()
		 */
		private void processRegisterMEResponse(WorkerData work) {
			RegisterMEResponse response = (RegisterMEResponse) work.getMessage();
			
			System.out.println(work.getServerName() + " sent " + response);
				
			if(response.getResponse()==true){
				System.out.println("responseOK");
					
				if(writeMEState.equals(WANTED) && lastMEReq.isEqual(response.getReqTime())){
					// we've collected a result from 1 server for write mutual exclusion
					System.out.println("vote");
						
					if(!registrationRequests.isEmpty()){
						System.out.println("registrationRequests NOT empty=" + registrationRequests.size());
						UserRegistrationRecord r = registrationRequests.get(0);
						System.out.println(r);
						
						if(r.getRequestTime()!=null && r.getRequestTime().equals(lastMEReq)){
							r.addAgreedServer(work.getServerName());
							
							if(allLiveServers(r.getAgreedServers())){
								// obtained the write ME
								writeMEState = HELD;
								System.out.println("allLiveServers:" + r.getAgreedServers());
								System.out.println("write mutual exclusion obtained for " + r);
									
								// multicast write command to all servers
								logicalClock.timestampNextEvent();
								WriteCommand command = new WriteCommand(r.getUsername(), r.getPassword(), logicalClock.getTimeStamp());
								command.setBytes(r.getBytes());
								command.setBG(r.getBG());
								command.setNickname(r.getNickname());
								work.getServer().multicastToServers(command);
									
								// write into local database
								DroidChatUserBean newUser = new DroidChatUserBean();
								newUser.setUsername(r.getUsername());
								newUser.setPassword(r.getPassword());
								newUser.setBytes(r.getBytes()); // new set picture
								newUser.setAvailability(Availability.AVAILABLE.ordinal());
								newUser.setBG(r.getBG());
								newUser.setNickname(r.getNickname());
								
								DroidChatUserBean createdUser = null;
									
								try {
									createdUser = userDAO.create(newUser);
								} catch (DAOException e) {	e.printStackTrace(); }
								
								System.out.println("written into database:" + createdUser);
									
								ClientRegisterResponseMessage responseToClient = 
									new ClientRegisterResponseMessage(ClientRegisterResponseMessage.REGISTER_SUCCESSFUL,
											createdUser.getUsername());
									
								// reply to client that command
								work.getServer().sendToClient(r.getSocket(), responseToClient);
								registrationRequests.remove(0); // done with this registration request
								
								// now that I'm released 
								writeMEState = RELEASED;
								
								// was there any pending MErequests for me to process?
								synchronized (holdbackQueue) {
									System.out.println("holdback queue size: " + holdbackQueue.size());
									
									if(!holdbackQueue.isEmpty()){
										RegisterMERequest req = (RegisterMERequest) holdbackQueue.get(0).getMessage();
										if(canProcess(req)){
											System.out.println(holdbackQueue.get(0).getMessage() + " can be processed now");
											synchronized(queue){
												queue.add(holdbackQueue.remove(0));
											}
										}
									} else {
										// was there any new RegisterRequest that I can now send out
										synchronized (registrationRequests) {
											if(!registrationRequests.isEmpty() && registrationRequests.get(0).getRequestTime()==null){
												// registration request that could not be multicasted is now being processed
												System.out.println("	registration request that could not be multicasted is now being processed");
												System.out.println("	before="+registrationRequests.get(0));
												MEClock.timestampNextEvent();
												logicalClock.timestampNextEvent();
												RegisterMERequest registerReq = new RegisterMERequest(MEClock.getTimeStamp(), r.getUsername(), logicalClock.getTimeStamp());
												lastMEReq = logicalClock.getTimeStamp();
												work.getServer().multicastToServers(registerReq);
												writeMEState = WANTED;
												System.out.println("	ME state=WANTED");
												registrationRequests.get(0).setRequestTime(lastMEReq);
												System.out.println("	after="+registrationRequests.get(0));
											}
										}
									}
								}
							} else {
								System.out.println("NOT allLiveServers:" + r.getAgreedServers());
							}
						}
					} else {
						System.out.println("registrationRequests empty");
					}
				}
			} else {
				System.out.println("responseNotOK");
			
				if(!registrationRequests.isEmpty()){
					System.out.println("registrationRequests NOT empty=" + registrationRequests.size());
					UserRegistrationRecord r = registrationRequests.get(0);
					
					if(r.getRequestTime()!=null && r.getRequestTime().equals(response.getReqTime())){
						// username has already been registered -> registration fails
						ClientRegisterResponseMessage responseToClient = new ClientRegisterResponseMessage(ClientRegisterResponseMessage.USERNAME_ALREADY_REGISTERED, r.getUsername());
						work.getServer().sendToClient(r.getSocket(), responseToClient);
						
						System.out.println("	matched with registrationRequests head=" + r);
						// now that I'm released
						writeMEState = RELEASED;
						registrationRequests.remove(0);
						System.out.println("	writeMEState is now " + writeMEState);
						System.out.println("	registrationRequests head removed=" + registrationRequests.size());
						
						
						// was there any pending MErequests for me to process?
						synchronized (holdbackQueue) {
							System.out.println("holdback queue size: " + holdbackQueue.size());
							
							if(!holdbackQueue.isEmpty()){
								RegisterMERequest req = (RegisterMERequest) holdbackQueue.get(0).getMessage();
								if(canProcess(req)){
									System.out.println(holdbackQueue.get(0).getMessage() + " can be processed now");
									synchronized(queue){
										queue.add(holdbackQueue.remove(0));
									}
								}
							} else {
								// was there any new RegisterRequest that I can now send out
								synchronized (registrationRequests) {
									if(!registrationRequests.isEmpty() && registrationRequests.get(0).getRequestTime()==null){
										// registration request that could not be multicasted is now being processed
										System.out.println("	registration request that could not be multicasted is now being processed");
										System.out.println("	before="+registrationRequests.get(0));
										MEClock.timestampNextEvent();
										logicalClock.timestampNextEvent();
										RegisterMERequest registerReq = new RegisterMERequest(MEClock.getTimeStamp(), r.getUsername(), logicalClock.getTimeStamp());
										lastMEReq = logicalClock.getTimeStamp();
										work.getServer().multicastToServers(registerReq);
										writeMEState = WANTED;
										System.out.println("	ME state=WANTED");
										registrationRequests.get(0).setRequestTime(lastMEReq);
										System.out.println("	after="+registrationRequests.get(0));
									}
								}
							}
						}
					}
				} else {
					System.out.println("registrationRequests empty");
				}
			}
		}
		
		/**
		 * worker thread helper method: processRegisterMERequest()
		 */
		private void processWriteCommand(WorkerData work) {
			WriteCommand writeCommand = (WriteCommand) work.getMessage();
			System.out.println(writeCommand + " from " + work.getServerName());
			
			if(writeMEState.equals(HELD)){
				System.out.println("THIS SHOULD NOT HAPPEN");
				System.exit(0);
			}
			
			DroidChatUserBean newUser = new DroidChatUserBean();
			newUser.setUsername(writeCommand.getUsername());
			newUser.setPassword(writeCommand.getPassword());
			newUser.setBytes(writeCommand.getBytes());
			newUser.setAvailability(Availability.AVAILABLE.ordinal());
			newUser.setBG(writeCommand.getBG());
			newUser.setNickname(writeCommand.getNickname());
			
			DroidChatUserBean createdUser = null;
			try {
				createdUser = userDAO.create(newUser);
			} catch (DAOException e) {	e.printStackTrace(); }
		}
		
		/**
		 *  For write Mutual Exclusion
		 *  does the list of agreedServers contain all servers that are
		 *  currently alive?
		 */  
		private boolean allLiveServers(HashSet<String> agreedServers) {
			boolean result = false;
			
			synchronized (serverInfo) {
				if(agreedServers.containsAll(serverInfo.keySet())){
					result = true;
				}
			}
			
			return result;
		}
	}
	
	// constructor
	public NioServer(String configuration_filename, String ServerName) throws IOException {
		this.myServerName = ServerName;
		properties = new Properties();
	    ReadConfigFile(configuration_filename, properties); // read the config file
	    initializeServerInfo();
	    this.selector = this.initSelector();
	    
	    String jdbcDriver = "com.mysql.jdbc.Driver";
	    String jdbcURL    = "jdbc:mysql:///DroidChat";
	    String userName   = null; // MySQL database userName null 
	    String password   = null; // MySQL database password null 

	    BeanTable.useJDBC(jdbcDriver,jdbcURL,userName,password);
	    
		try {
			userDAO = new DroidChatUserDAO();
		} catch (DAOException e2) { e2.printStackTrace(); }
		
		changeRequests = new LinkedList<ChangeRequest>();
		pendingData = new HashMap<SelectableChannel, LinkedList<ByteBuffer>>();
		
		pendingIncomingData = new HashMap<SocketChannel, LinkedList<ByteBuffer>>();
		
		userLoginRecord = new HashMap<Integer, Login_Session_Attr>();
		registrationRequests = new ArrayList<UserRegistrationRecord>();
		
		UDPchannel = DatagramChannel.open(); 
		UDPsocket = UDPchannel.socket();
		UDPsocket.bind(new InetSocketAddress (udp_port));
		UDPchannel.configureBlocking(false);
		UDPchannel.register(selector, SelectionKey.OP_READ);
		
		// creates fake logged in user
		/*Login_Session_Attr attr = new Login_Session_Attr(12345, "peter14f");
		attr.setLast_known_lattitude(fake_latitude);
		attr.setLast_known_longitude(fake_longitude);
		attr.setIP(myAddress);
		attr.setRound(1);
		
		userLoginRecord.put(12345, attr);*/
		
		worker = new ClientMessageWorker(userDAO);
		holdbackQueue = new ArrayList<WorkerData>();
		new Thread(worker).start();
	}
	
	/**
	 * This method is for accepting a connection request 
	 *  coming from another server
	 * 
	 * return SocketChannel if the connection is indeed coming from an earlier client
	 * return null otherwise
	 */
	private SocketChannel acceptAnotherServer(SelectionKey key) throws IOException {
		ServerSocketChannel serverSocketChannel = (ServerSocketChannel) key.channel();
		
		SocketChannel socketChannel = serverSocketChannel.accept();
		socketChannel.configureBlocking(false);	
		
		InetAddress addr = socketChannel.socket().getInetAddress();
		String address = addr.getHostAddress();
		
		Server s = lookupServerByIP(address);
		
		if(s==null){
			socketChannel.close();
			return null;
		}
		
		// register the new channel only if I am the "later server"
		if(s.getServerName().compareTo(myServerName) < 0){
			// Register the new SocketChannel with our Selector, indicating
			// we'd like to be notified when there's data waiting to be read

			if(!s.isChannelEstablished()){
				// channel not established yet
				s.setChannel(socketChannel);
				s.setConnectionEstablished(true);
				System.out.println("registering socketChannel - a connection from an earlier server: " + s.getServerName());
				socketChannel.register(selector, SelectionKey.OP_READ);
			}else{
				// channel already established
				System.out.println("channel already established - why did he make two connections");
				System.exit(0);
			}
			
			return socketChannel;
		}else{
			return null;
		}
	}
	
	
	/**
	 * This method is used to accept client connections from DroidChat users
	 * 
	 * @param key
	 * @throws IOException
	 */
	private void accept(SelectionKey key) throws IOException {
		ServerSocketChannel serverSocketChannel = (ServerSocketChannel) key.channel();
		
		SocketChannel socketChannel = serverSocketChannel.accept();
		socketChannel.configureBlocking(false);	
		
		// Register the new SocketChannel with our Selector, indicating
	    // we'd like to be notified when there's data waiting to be read
		System.out.println("registering READ interest in socketChannel" + socketChannel);
	    socketChannel.register(selector, SelectionKey.OP_READ);
	    socketChannel.socket().setTcpNoDelay(true);
	}

	/**
	 *  initialize the two listening sockets, one for other servers, and one
	 *   for clients and returns the selector
	 */
	private Selector initSelector() throws IOException {
		// Create a new selector
		Selector socketSelector = SelectorProvider.provider().openSelector();

		/*** serverChannel1 is for listening to other server connections ***/
		// Create a new non-blocking server socket channel
		serverChannel1 = ServerSocketChannel.open();
		serverChannel1.configureBlocking(false);

		// Bind the server socket to the specified address and port
		InetSocketAddress isa = new InetSocketAddress(myAddress, port1);
		serverChannel1.socket().bind(isa);
		
		// Register the server socket channel, indicating an interest in accepting new connections
		serverChannel1.register(socketSelector, SelectionKey.OP_ACCEPT);
		
		/*** serverChannel2 is for listening to client connections ***/
		// Create a new non-blocking server socket channel
		serverChannel2 = ServerSocketChannel.open();
		serverChannel2.configureBlocking(false);
		
		// Bind the server socket to the specified address and port
		InetSocketAddress isa2 = new InetSocketAddress(myAddress, port2);
		serverChannel2.socket().bind(isa2);
		
		// Register the server socket channel, indicating an interest in accepting new connections
		serverChannel2.register(socketSelector, SelectionKey.OP_ACCEPT);
		
		return socketSelector;
	}
	
	/**
	 * Call this function when a server's channel has data for you to read
	 * the selecting thread calls this, when select detects a channel is readable
	 *  socketChannel.read() is called here
	 * 
	 * @param key
	 * @throws IOException
	 */
	private void readFromServer(SelectionKey key, Server s) throws IOException {
		System.out.println("	readFromServer()");
		
		SocketChannel socketChannel = (SocketChannel) key.channel();
		
		int numRead;
		
		synchronized (pendingIncomingData) {
			LinkedList<ByteBuffer> IncomingBufferForChannel = pendingIncomingData.get(socketChannel);
			
			if (IncomingBufferForChannel == null) {
				IncomingBufferForChannel = new LinkedList<ByteBuffer>();
				pendingIncomingData.put(socketChannel, IncomingBufferForChannel);
				
				IncomingBufferForChannel.add(ByteBuffer.allocate(4)); // [0] is the byteSize of the message
				
				try {
					numRead = socketChannel.read(IncomingBufferForChannel.get(0));
				} catch (IOException e) {
					// The remote forcibly closed the connection, cancel
					// the selection key and close the channel.
					closeChannelWithServer(key, s);
					return;
				}
				if (numRead == -1) {
					closeChannelWithServer(key, s);
					return;
				} else if (numRead==4) {
					IncomingBufferForChannel.get(0).flip();
					int byteSize = IncomingBufferForChannel.get(0).getInt();
					System.out.println(numRead + " bytes read from server=" + s.getServerName() + ". byteSize:" + byteSize);
					
					IncomingBufferForChannel.add(ByteBuffer.allocate(byteSize)); // [1] is the actual message
					ByteBuffer actualMsg = IncomingBufferForChannel.get(1);
					
					try {
						numRead = socketChannel.read(actualMsg);
					} catch (IOException e) {
						closeChannelWithServer(key, s);
						return;
					}
					
					if (numRead == -1) {
						// Remote entity shut the socket down cleanly. Do the
					    // same from our end and cancel the channel.
						closeChannelWithServer(key, s);
						return;
					}else{
						System.out.println(numRead + " bytes read from server " + s.getServerName());
						if(actualMsg.remaining()==0){
							ByteArrayInputStream bais = new ByteArrayInputStream(actualMsg.array());
							ObjectInputStream ois = new ObjectInputStream(bais);
							ServerMessage m = null;
							try {
								m = (ServerMessage) ois.readObject();
							} catch (ClassNotFoundException e) {e.printStackTrace();}
							
							IncomingBufferForChannel.clear();
							pendingIncomingData.remove(socketChannel);
							
							if (m instanceof RegisterMERequest) {
								System.out.println("	A RegisterMERequest");
								/** RegisterMERequest message **/ 
								RegisterMERequest MEReq = (RegisterMERequest) m;
								if (canProcess(MEReq)) {
									// hand the ServerMessage off to the worker thread
									//  the selecting thread calls this method - processData
									//  which wakes up the worker thread
									this.worker.processReceivedServerMessage(this, socketChannel, m, s);
								} else {
									// put it in the holdback queue for now
									this.worker.insertRegisterMERequest(this, socketChannel, MEReq, s);
								}
							} else {
								/** unicast message **/
								if (m instanceof UserLoginAnnouncement) {
									System.out.println("	A UserLoginAnnouncement");
								} else if (m instanceof RegisterMEResponse){
									System.out.println("	A RegisterMEResponse");
								} else if (m instanceof WriteCommand){
									System.out.println("	A WriteCommand");
								} else if (m instanceof FindNeighborMessage){
									System.out.println("	A FindNeighborMessage");
								} else {
									System.out.println("	A WHAT?!");
								} 
								
								// hand the ServerMessage off to the worker thread
								//  the selecting thread calls this method - processData
								//  which wakes up the worker thread
								this.worker.processReceivedServerMessage(this, socketChannel, m, s);
							}
						}
					}
				}
			} else {
				ByteBuffer byteSizeBuf = IncomingBufferForChannel.get(0);
				int remaining = byteSizeBuf.remaining();
				
				if(remaining==0){
					// I already know how many bytes the message is
					ByteBuffer actualMsg = IncomingBufferForChannel.get(1);
					
					try {
						numRead = socketChannel.read(actualMsg);
					} catch (IOException e) {
						closeChannelWithServer(key, s);
						return;
					}
					if (numRead == -1) {
						closeChannelWithServer(key, s);
					} else {
						System.out.println(numRead + " bytes read from server:" + s.getServerName());
						if(actualMsg.remaining()==0){
							ByteArrayInputStream bais = new ByteArrayInputStream(actualMsg.array());
							ObjectInputStream ois = new ObjectInputStream(bais);
							ServerMessage m = null;
							try {
								m = (ServerMessage) ois.readObject();
							} catch (ClassNotFoundException e) {e.printStackTrace();}
							
							IncomingBufferForChannel.clear();
							pendingIncomingData.remove(socketChannel);
							
							if (m instanceof RegisterMERequest) {
								System.out.println("	A RegisterMERequest");
								/** RegisterMERequest message **/ 
								RegisterMERequest MEReq = (RegisterMERequest) m;
								if (canProcess(MEReq)) {
									// hand the ServerMessage off to the worker thread
									//  the selecting thread calls this method - processData
									//  which wakes up the worker thread
									this.worker.processReceivedServerMessage(this, socketChannel, m, s);
								} else {
									// put it in the holdback queue for now
									this.worker.insertRegisterMERequest(this, socketChannel, MEReq, s);
								}
							} else {
								/** unicast message **/
								if (m instanceof UserLoginAnnouncement) {
									System.out.println("	A UserLoginAnnouncement");
									System.out.println("Server" + s + " " + (UserLoginAnnouncement)m);
								} else if (m instanceof RegisterMEResponse){
									System.out.println("	A RegisterMEResponse");
								} else if (m instanceof WriteCommand){
									System.out.println("	A WriteCommand");
								} else if (m instanceof FindNeighborMessage){
									System.out.println("	A FindNeighborMessage");
								} else {
									System.out.println("	A WHAT?!");
								}
								// hand the ServerMessage off to the worker thread
								//  the selecting thread calls this method - processData
								//  which wakes up the worker thread
								this.worker.processReceivedServerMessage(this, socketChannel, m, s);
							}
						}
					}
				}
			}
		}
	}
	
	
	private void closeChannelWithServer(SelectionKey key, Server s){
		SocketChannel socketChannel = (SocketChannel) key.channel();
		key.cancel(); // cancel the registration of this key with its selector
		try {
			socketChannel.close();
		} catch (IOException e) {e.printStackTrace();}
		synchronized (changeRequests) {
			Iterator<ChangeRequest> itr = changeRequests.iterator();
			while(itr.hasNext()){
				ChangeRequest r = itr.next();
				if(r.socket==socketChannel){
					itr.remove();
					System.out.println("		removed " + socketChannel + " from changeRequests");
				}
			}
		}
		pendingIncomingData.remove(socketChannel);
		deadServer.push(s.getServerName());
		System.out.println("	server " + s.getServerName() + " died");
	}
	
	/**
	 * Should I be processing this RegisterMERequest?
	 *  deliverable && REALASED
	 * 	  or
	 *  deliverable && WANTED && a smaller METime
	 * 
	 */
	private boolean canProcess(RegisterMERequest MEReq) {
		LogicalTimeStamp reqMsgTime = MEReq.getMessageTS();
		
		if (deliverable(MEReq)) {
			if (writeMEState.equals(RELEASED)) {
				return true;
			} else if (writeMEState.equals(WANTED)) {
				if(reqMsgTime.happenedBefore(this.lastMEReq)){
					// I lose
					return true;
				}else if(reqMsgTime.isEqual(this.lastMEReq)){
					int rPos = reqMsgTime.getPos();
					if(rPos < this.myPos){
						// I lose
						return true;
					}else{
						// I win
						return false; // don't process it yet
					}
				}else{
					/** A later request than mine **/
					return false; // don't process it yet
				}
			}
		} else {
			return false; // don't process it yet
		}
		return false;
	}

	/**
	 * returns true if RegisterMERequest m can be delivered
	 *  according to the Causal Ordering rule
	 */
	private boolean deliverable(RegisterMERequest m) {
		boolean result = true;
		
		synchronized(MEClock){
			VectorTimeStamp current = MEClock.getTimeStamp();
			VectorTimeStamp message_vectorTS = m.getME_TS();
		
			int msgPos = message_vectorTS.getPos();
		
			int[] current_ts = current.getVectorTimeStamp();
			int[] message_ts = message_vectorTS.getVectorTimeStamp();
		
			if (message_ts[msgPos] == current_ts[msgPos] + 1) {
				for (int i = 0; i< current_ts.length; i++) {
					if (i != msgPos) {
						if (message_ts[i] > current_ts[i])
							result = false;
					}
				}
			} else {
				result = false;
			}
		}
		return result;
	}

	private void printServersLeft() {
		
		System.out.println("servers left: ");
		
		synchronized (serverInfo) {
			for(String s: serverInfo.keySet()){
				System.out.println("	" + serverInfo.get(s));
			}
		}
	}

	/**
	 * Call this function when a client's channel has data for you to read
	 * the selecting thread calls this, when select detects a channel is readable
	 *  socketChannel.read() is called here
	 * 
	 * @param key
	 * @throws IOException
	 */
	private void readFromClient(SelectionKey key) throws IOException {
		//System.out.println("	readFromClient()");
		
		SocketChannel socketChannel = (SocketChannel) key.channel();
	
		int numRead;
		
		synchronized (pendingIncomingData) {
			LinkedList<ByteBuffer> IncomingBufferForChannel = pendingIncomingData.get(socketChannel);
			
			if (IncomingBufferForChannel == null) {
				IncomingBufferForChannel = new LinkedList<ByteBuffer>();
				pendingIncomingData.put(socketChannel, IncomingBufferForChannel);
				
				IncomingBufferForChannel.add(ByteBuffer.allocate(4)); // [0] is the byteSize of the message
				
				try {
					numRead = socketChannel.read(IncomingBufferForChannel.get(0));
				} catch (IOException e) {
					// The remote forcibly closed the connection, cancel
					// the selection key and close the channel.
					closeChannelWithClient(key);
					return;
				}
				if (numRead == -1) {
					// Remote entity shut the socket down cleanly. Do the
				    // same from our end and cancel the channel.
					closeChannelWithClient(key);
					return;
				} else if (numRead==4) {
					IncomingBufferForChannel.get(0).flip();
					
					int byteSize = IncomingBufferForChannel.get(0).getInt();
					
					//System.out.println(numRead + " bytes read from client. byteSize:" + byteSize);
					
					IncomingBufferForChannel.add(ByteBuffer.allocate(byteSize)); // [1] is the actual message
					ByteBuffer actualMsg = IncomingBufferForChannel.get(1);
					
					try {
						numRead = socketChannel.read(actualMsg);
					} catch (IOException e) {
						// The remote forcibly closed the connection, cancel
						// the selection key and close the channel.
						closeChannelWithClient(key);
						return;
					}
					
					if (numRead == -1) {
						// Remote entity shut the socket down cleanly. Do the
					    // same from our end and cancel the channel.
						closeChannelWithClient(key);
						return;
					}else{
						//System.out.println(numRead + " bytes read from client:");
						if(actualMsg.remaining()==0){
							ByteArrayInputStream bais = new ByteArrayInputStream(actualMsg.array());
							ObjectInputStream ois = new ObjectInputStream(bais);
							DroidChatMessage m = null;
							try {
								m = (DroidChatMessage) ois.readObject();
							} catch (ClassNotFoundException e) {e.printStackTrace();}
							
							if(m instanceof ClientLoginMessage){
								System.out.println("	received a ClientLoginMessage: " +(ClientLoginMessage)m);
							} else if(m instanceof ClientRegisterMessage){
								System.out.println("	received a ClientRegisterMessage: " +(ClientRegisterMessage)m);
							} else if(m instanceof ClientUpdateLocationMessage){
								System.out.println("	received a ClientUpdateLocationMessage: " +(ClientUpdateLocationMessage)m);
							}
							
							// Hand the DroidChatMessage off to our worker thread
							//  the selecting thread calls this method, which wakes up the 
							//   worker thread
							this.worker.processReceivedDroidChatMessage(this, socketChannel, m);
							pendingIncomingData.remove(socketChannel);
						}
					}
				}
			} else {
				ByteBuffer byteSizeBuf = IncomingBufferForChannel.get(0);
				int remaining = byteSizeBuf.remaining();
				
				if(remaining==0){
					// I already know how many bytes the message is
					ByteBuffer actualMsg = IncomingBufferForChannel.get(1);
					
					try {
						numRead = socketChannel.read(actualMsg);
					} catch (IOException e) {
						// The remote forcibly closed the connection, cancel
						// the selection key and close the channel.
						closeChannelWithClient(key);
						return;
					}
					
					if (numRead == -1) {
						// Remote entity shut the socket down cleanly. Do the
					    // same from our end and cancel the channel.
						closeChannelWithClient(key);
						return;
					}else{
						System.out.println(numRead + " bytes read from client:");
						if(actualMsg.remaining()==0){
							ByteArrayInputStream bais = new ByteArrayInputStream(actualMsg.array());
							ObjectInputStream ois = new ObjectInputStream(bais);
							DroidChatMessage m = null;
							try {
								m = (DroidChatMessage) ois.readObject();
							} catch (ClassNotFoundException e) {e.printStackTrace();}
							
							if(m instanceof ClientLoginMessage){
								System.out.println("	received a ClientLoginMessage: " +(ClientLoginMessage)m);
							}else if(m instanceof ClientRegisterMessage){
								System.out.println("	received a ClientRegisterMessage: " +(ClientRegisterMessage)m);
							}else if(m instanceof ClientUpdateLocationMessage){
								System.out.println("	received a ClientUpdateLocationMessage: " +(ClientUpdateLocationMessage)m);
							}
							
							// Hand the DroidChatMessage off to our worker thread
							//  the selecting thread calls this method, which wakes up the 
							//   worker thread
							this.worker.processReceivedDroidChatMessage(this, socketChannel, m);
							pendingIncomingData.remove(socketChannel);
						}
					}
				} else {
					try {
						numRead = socketChannel.read(byteSizeBuf);
					} catch (IOException e) {
						// The remote forcibly closed the connection, cancel
						// the selection key and close the channel.
						closeChannelWithClient(key);
						return;
					}
					if (numRead == -1) {
						// Remote entity shut the socket down cleanly. Do the
					    // same from our end and cancel the channel.
						closeChannelWithClient(key);
						return;
					} else if (numRead==remaining) {
						byteSizeBuf.flip();
						int byteSize = byteSizeBuf.getInt();
						
						// finished reading in the byteSize
						System.out.println("byteSize:" + byteSize);
						
						IncomingBufferForChannel.add(ByteBuffer.allocate(byteSize)); // [1] is the actual message
						ByteBuffer actualMsg = IncomingBufferForChannel.get(1);
						
						try {
							numRead = socketChannel.read(actualMsg);
						} catch (IOException e) {
							// The remote forcibly closed the connection, cancel
							// the selection key and close the channel.
							closeChannelWithClient(key);
							return;
						}
						
						if (numRead == -1) {
							// Remote entity shut the socket down cleanly. Do the
						    // same from our end and cancel the channel.
							closeChannelWithClient(key);
							return;
						}else{
							System.out.println(numRead + " bytes read from client:");
							if(actualMsg.remaining()==0){
								ByteArrayInputStream bais = new ByteArrayInputStream(actualMsg.array());
								ObjectInputStream ois = new ObjectInputStream(bais);
								DroidChatMessage m = null;
								try {
									m = (DroidChatMessage) ois.readObject();
								} catch (ClassNotFoundException e) {e.printStackTrace();}
								
								if(m instanceof ClientLoginMessage){
									System.out.println("	received a ClientLoginMessage: " +(ClientLoginMessage)m);
								}else if(m instanceof ClientRegisterMessage){
									System.out.println("	received a ClientRegisterMessage: " +(ClientRegisterMessage)m);
								}
								
								// Hand the DroidChatMessage off to our worker thread
								//  the selecting thread calls this method, which wakes up the 
								//   worker thread
								this.worker.processReceivedDroidChatMessage(this, socketChannel, m);
								pendingIncomingData.remove(socketChannel);
							}
						}
					}
				}
			}
		}
	}

	private void closeChannelWithClient(SelectionKey key) {
		SocketChannel socketChannel = (SocketChannel) key.channel();
		key.cancel();
		try {
			socketChannel.close();
		} catch (IOException e) {e.printStackTrace();}
		synchronized (changeRequests) {
			Iterator<ChangeRequest> itr = changeRequests.iterator();
			while(itr.hasNext()){
				ChangeRequest r = itr.next();
				if(r.socket==socketChannel){
					itr.remove();
					System.out.println("		removed " + socketChannel + " from changeRequests");
				}
			}
		}
		pendingIncomingData.remove(socketChannel);
	}

	private String intToIp(int i) {
        return ((i >> 24 ) & 0xFF) + "." +
               ((i >> 16 ) & 0xFF) + "." +
               ((i >>  8 ) & 0xFF) + "." +
               ( i        & 0xFF);
    }
	
	/**
	 * Call this function when there is data to be written into a DatagramChannel
	 *  and that channel has has been confirmed writable
	 * 
	 * This method just has to pull DroidChatMessage off the appropriate queue and
	 *  keep writing it until it either runs out or can't write anymore.
	 *  
	 * the selecting thread calls this, when select detects channel is writable
	 * 
	 * socketChannel.write()
	 * 
	 * @param key
	 * @throws IOException 
	 */
	private void writeUDP(SelectionKey key) throws IOException {
		System.out.println("writeUDP()");
		
		DatagramChannel datagramChannel = (DatagramChannel) key.channel();
		
		synchronized (pendingData) {
			LinkedList<ByteBuffer> messagedToSend = pendingData.get(datagramChannel);
			
			//datagramChannel.
			
			// Write until there's not more data ...
			while (!messagedToSend.isEmpty()) {
				ByteBuffer dest = messagedToSend.get(0);
				ByteBuffer data = messagedToSend.get(1);
				
				String destination_ip = intToIp((int)dest.getLong());
				InetSocketAddress target = new InetSocketAddress(destination_ip, client_udpport);
				
				//System.out.println(" writing " + data.capacity());
				int n = datagramChannel.send(data, target);
				System.out.println(" " + n + " bytes wrote");
				
				if (data.remaining() > 0) {
					// ... or the socket's buffer fills up
					System.out.println(" socket's buffer fills up");
					break;
				}
				//System.out.println(" removing ");
				messagedToSend.remove();
				messagedToSend.remove();
				
				//messagedToSend.remove(0);
			}
			
			if (messagedToSend.isEmpty()) {
				// We wrote away all data, so we're no longer interested
		        // in writing on this socket. Switch back to waiting for data.
		        key.interestOps(SelectionKey.OP_READ);
			}
		}
	}
	
	/**
	 * Call this function when there is data to be written into a SocketChannel
	 *  and that channel has has been confirmed writable
	 * 
	 * This method just has to pull DroidChatMessage off the appropriate queue and
	 *  keep writing it until it either runs out or can't write anymore.
	 *  
	 * the selecting thread calls this, when select detects channel is writable
	 * 
	 * socketChannel.write()
	 * 
	 * @param key
	 * @throws IOException 
	 */
	private void write(SelectionKey key) throws IOException {
		SocketChannel socketChannel = (SocketChannel) key.channel();
		
		synchronized (pendingData) {
			LinkedList<ByteBuffer> messagedToSend = pendingData.get(socketChannel);
			
			// Write until there's not more data ...
			while (!messagedToSend.isEmpty()) {
				ByteBuffer buf = messagedToSend.get(0);
				//System.out.println(" writing " + buf.capacity());
				socketChannel.write(buf);
				if (buf.remaining() > 0) {
					// ... or the socket's buffer fills up
					System.out.println(" socket's buffer fills up");
					break;
				}
				//System.out.println(" removing ");
				messagedToSend.remove(0);
			}
			
			if (messagedToSend.isEmpty()) {
				// We wrote away all data, so we're no longer interested
		        // in writing on this socket. Switch back to waiting for data.
		        key.interestOps(SelectionKey.OP_READ);
			}
		}
	}
	
	/*
	 *  the worker thread calls this method to queue up multicast message that
	 *    needs to be sent out by the selecting thread to all servers
	 *    wakes up the selecting thread at the end
	 */
	public void multicastToServers(ServerMessage m) {
		
		System.out.println("	multicastToServers()");
		
		synchronized (serverInfo) {
			for(String s: serverInfo.keySet()){
				synchronized (changeRequests) {
					Server server = serverInfo.get(s);
					SocketChannel socket = server.getChannel();
					// Indicate we want the interest ops set to change for socket
					changeRequests.add(new ChangeRequest(socket, ChangeRequest.CHANGEOPS, SelectionKey.OP_WRITE));
					
					ByteArrayOutputStream baos = new ByteArrayOutputStream();
					ObjectOutputStream oos = null;
					try {
						oos = new ObjectOutputStream(baos);
						oos.writeObject(m);
						oos.flush();
					} catch (IOException e) {
						e.printStackTrace();
						System.out.println("problem");
						return;
					}
					
					ByteBuffer header = ByteBuffer.allocate(4);
					int byteSize = baos.toByteArray().length;
					header.putInt(byteSize);
					
					// And queue the ServerMessage we want written
					synchronized (pendingData) {
						LinkedList<ByteBuffer> MessageQueueForChannel = pendingData.get(socket);
						if (MessageQueueForChannel == null) {
							MessageQueueForChannel = new LinkedList<ByteBuffer>();
							pendingData.put(socket, MessageQueueForChannel);
						}
						MessageQueueForChannel.add(ByteBuffer.wrap(header.array()));
						MessageQueueForChannel.add(ByteBuffer.wrap(baos.toByteArray()));
						System.out.println("ServerMessage " + m + " added in queue is " + byteSize + " bytes long");
					}
				}
			}
		}
	}
	
	// the worker thread calls this method to queue up messages that
	//  needs to be sent out by the selecting thread to a server (specified by socket) 
	//  wakes up the selecting thread at the end
	public void sendToServer(SocketChannel socket, ServerMessage message) {
		System.out.println("	sendToServer()");
		
		synchronized (changeRequests) {
			// Indicate we want the interest ops set to change for socket
			changeRequests.add(new ChangeRequest(socket, ChangeRequest.CHANGEOPS, SelectionKey.OP_WRITE));
			
			ByteArrayOutputStream baos = new ByteArrayOutputStream();
			ObjectOutputStream oos = null;
			try {
				oos = new ObjectOutputStream(baos);
				oos.writeObject(message);
				oos.flush();
			} catch (IOException e) {
				e.printStackTrace();
				System.out.println("problem");
				return;
			}
			
			ByteBuffer header = ByteBuffer.allocate(4);
			int byteSize = baos.toByteArray().length;
			header.putInt(byteSize);
			
			// And queue the DroidChatMessage we want written
			synchronized (pendingData) {
				LinkedList<ByteBuffer> MessageQueueForChannel = pendingData.get(socket);
				if (MessageQueueForChannel == null) {
					MessageQueueForChannel = new LinkedList<ByteBuffer>();
					pendingData.put(socket, MessageQueueForChannel);
				}
				MessageQueueForChannel.add(ByteBuffer.wrap(header.array()));
				MessageQueueForChannel.add(ByteBuffer.wrap(baos.toByteArray()));
				System.out.println("ServerMessage" + message + " added in queue");
			}
		}
		
		//wake up the selecting thread so it can make the required changes
	    selector.wakeup(); // causes select() to return
	}
	
	/**
	 * a simple helper method translate IP in the "xxx.xxx.xxx.xxx" format
	 *   into a 8-byte long
	 */
	private long ipToInt(String addr) {
        String[] addrArray = addr.split("\\.");

        long num = 0;
        for (int i=0;i<addrArray.length;i++) {
            int power = 3-i;

            num += ((Integer.parseInt(addrArray[i])%256 * Math.pow(256,power)));
        }
        return num;
    }
	
	public void sendToClientUDP(DroidChatMessage message, String destination_ip){
		synchronized (changeRequests) {
			// Indicate we want the interest ops set to change for socket
			changeRequests.add(new ChangeRequest(UDPchannel, ChangeRequest.CHANGEOPS, SelectionKey.OP_WRITE));
			
			ByteArrayOutputStream baos = new ByteArrayOutputStream();
			ObjectOutputStream oos = null;
			try {
				oos = new ObjectOutputStream(baos);
				oos.writeObject(message);
				oos.flush();
			} catch (IOException e) {
				e.printStackTrace();
				System.out.println("problem");
				return;
			}

			// And queue the DroidChatMessage we want written
			synchronized (pendingData) {
				LinkedList<ByteBuffer> MessageQueueForChannel = pendingData.get(UDPchannel);
				if (MessageQueueForChannel == null) {
					MessageQueueForChannel = new LinkedList<ByteBuffer>();
					pendingData.put(UDPchannel, MessageQueueForChannel);
				}
				ByteBuffer dest = ByteBuffer.allocate(8);
				dest.putLong(ipToInt(destination_ip));
				
				//int byteSize = baos.toByteArray().length;
				//dest.
				MessageQueueForChannel.add(ByteBuffer.wrap(dest.array()));
				MessageQueueForChannel.add(ByteBuffer.wrap(baos.toByteArray()));
				System.out.println("DroidChatMessage" + message + " added in queue (udpchannel): " );
			}
		}
		
		//wake up the selecting thread so it can make the required changes
	    selector.wakeup(); // causes select() to return
	}
	
	// the worker thread calls this method to queue up messages that
	//  needs to be sent out by the selecting thread to a client (specified by socket) 
	//  wakes up the selecting thread at the end
	public void sendToClient(SocketChannel socket, DroidChatMessage message) {
		synchronized (changeRequests) {
			// Indicate we want the interest ops set to change for socket
			changeRequests.add(new ChangeRequest(socket, ChangeRequest.CHANGEOPS, SelectionKey.OP_WRITE));
			
			ByteArrayOutputStream baos = new ByteArrayOutputStream();
			ObjectOutputStream oos = null;
			try {
				oos = new ObjectOutputStream(baos);
				oos.writeObject(message);
				oos.flush();
			} catch (IOException e) {
				e.printStackTrace();
				System.out.println("problem");
				return;
			}
			
			// And queue the DroidChatMessage we want written
			synchronized (pendingData) {
				LinkedList<ByteBuffer> MessageQueueForChannel = pendingData.get(socket);
				if (MessageQueueForChannel == null) {
					MessageQueueForChannel = new LinkedList<ByteBuffer>();
					pendingData.put(socket, MessageQueueForChannel);
				}
				ByteBuffer header = ByteBuffer.allocate(4);
				int byteSize = baos.toByteArray().length;
				header.putInt(byteSize);
				MessageQueueForChannel.add(ByteBuffer.wrap(header.array()));
				MessageQueueForChannel.add(ByteBuffer.wrap(baos.toByteArray()));
				System.out.println("DroidChatMessage" + message + " added in queue: " + byteSize + " bytes long");
			}
		}
		
		//wake up the selecting thread so it can make the required changes
	    selector.wakeup(); // causes select() to return
	}

	/**
	 *  open the config file and store attributes into prop
	 */
	private void ReadConfigFile(String configFile, Properties prop) {
		FileInputStream fis;
		try {
			fis = new FileInputStream(configFile);
			prop.clear();
			prop.load(fis);
			fis.close();
		} catch (FileNotFoundException e) {
			e.printStackTrace();
		}catch (IOException e) {
			e.printStackTrace();
		}
	}
	
	// called once at the beginning
	private void initializeServerInfo() {
		String temp = (String) properties.get("NAMES");
		System.out.println(temp);
		String[] name_list = temp.split(",");
		
		serverInfo = new HashMap<String, Server>();
		
		for(int i=0; i < name_list.length; i++){
			String serverName = name_list[i];
			if(!serverName.equals(myServerName)){
				Server newServer = new Server(serverName, properties.getProperty("server." + serverName + ".ip"), Integer.parseInt(properties.getProperty("server." + serverName + ".port1")));
				serverInfo.put(serverName, newServer);
				System.out.println(newServer);
			}else{
				myPos=i;
				port1 = Integer.parseInt(properties.getProperty("server." + serverName + ".port1"));
				port2 = Integer.parseInt(properties.getProperty("server." + serverName + ".port2"));
				myAddress = properties.getProperty("server." + myServerName + ".ip");
			}
		}
		deadServer = new Stack<String>();
		MEClock = new VectorClock(name_list.length, myPos);
		logicalClock = new LogicalClock(myPos);
	}
	
	private SocketChannel connectToServer(String ip, int port) throws IOException {
		// Create a non-blocking socket channel
		SocketChannel socketChannel = SocketChannel.open();
		socketChannel.configureBlocking(false);
		
		// Kick off connection establishment
		socketChannel.connect(new InetSocketAddress(ip, port));
		return socketChannel;
	}
	
	// called by the selecting thread itself
	private boolean finishConnection(SelectionKey key) throws IOException {
		SocketChannel socketChannel = (SocketChannel) key.channel();
	  
		boolean connected = false;
		
		// Finish the connection. If the connection operation failed
		// this will raise an IOException.
		try {
			connected = socketChannel.finishConnect();
		} catch (IOException e) {
			// Cancel the channel's registration with our selector
			key.cancel();
		}
		return connected;
	}
	
	/* 
	 * this is going to be the selecting thread in which select() will be called,
	 *  make sure that you don't call select() anywhere else!
	 */
	public void run() {
		while (true) {
	        try {
	        	// update the interest ops for the selection key on a given socket channel
	        	synchronized(changeRequests) {
	        		for(ChangeRequest change: changeRequests){
	        			switch(change.type) {
	        				case ChangeRequest.CHANGEOPS:
	        					SelectionKey key = change.socket.keyFor(selector);
	        					key.interestOps(change.ops);
	        			}
	        		}
	        		changeRequests.clear();
	        	}
	        	
	        	selector.select(2000);  // n is the number of selected-keys
	        	
				Set<SelectionKey> selectKeys = selector.selectedKeys();
				
				if(!selectKeys.isEmpty()){
					if(selectKeys.contains(serverChannel1.keyFor(selector))){
						// the "listen to other server" socket is now ready
						SelectionKey k = serverChannel1.keyFor(selector);
						
						if(k.isAcceptable()){
							SocketChannel newChannel = acceptAnotherServer(k);
							System.out.println(newChannel);
							// remove the key anyways
							selectKeys.remove(k);
						}
					}
				}
				
				// do I need to delete dead servers from serverInfo?
				deleteDeadServer();
				
				// do I need to connect to other servers?
				examineServers(selectKeys);

				for(SelectionKey key: selectKeys){
					if(key.isAcceptable()){
						// must be a connection from a DroidChat client
						System.out.println("acceptable");
						accept(key);
					}else if(key.isReadable()) {
						//System.out.println("readable");
						if(key.channel()==UDPchannel){
							System.out.println("	UDPchannel is ready for read");
							
							DatagramChannel ch = (DatagramChannel) key.channel();
							ByteBuffer temp = ByteBuffer.allocate(1024);
							SocketAddress sender = ch.receive(temp);
							
							if(sender==null){
								System.out.println(" ch.receive() returns null ");
							}else{
								System.out.println("	from " + sender.toString());
							}
						}else{
							readFromClient(key);
						}
					}else if(key.isWritable()){
						//System.out.println("writable");
						if(key.channel()==UDPchannel){
							writeUDP(key);
						}else{
							write(key);
						}
					}
				}
				selectKeys.clear();
				
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
		
	}

	/**
	 *  deadServer is a stack that contains server names
	 *   that were detected to have ended the TCP connection
	 */
	private void deleteDeadServer() {
		if(!deadServer.isEmpty()){
			while(!deadServer.isEmpty()){
				String sName = deadServer.pop();
				serverInfo.remove(sName);
				System.out.println("	server " + sName + " removed");
			}
			printServersLeft();
		}
	}

	/**
	 * for making connections to other servers
	 *  or detecting already established connections being dead
	 *  or reading data from other server
	 */
	private void examineServers(Set<SelectionKey> selectKeys) {
		try {
			synchronized(serverInfo){
				
				// iterate over the servers
				for(String serverName: serverInfo.keySet()){
					Server s = serverInfo.get(serverName);
				
					if(s.hasNoChannel()){
						System.out.println("	don't have a connection with server " + serverName + " yet");
					
						// is this a "later server"
						if(serverName.compareTo(myServerName) > 0){
							// a later server
							// I'm gonna initialize the connection
							SocketChannel newChannel;
						
							newChannel = connectToServer(s.getIp(), s.getPort());
						
							System.out.println("connecting to " + s.getIp() + " " + s.getPort());
							s.setChannel(newChannel);
							s.setConnectionAttempted(true);
							newChannel.register(selector, SelectionKey.OP_CONNECT);
						} // else he is supposed to connect to me, i just wait
					}else if(!s.isChannelEstablished()){
					
						if(serverName.compareTo(myServerName) > 0){
							// a later server
							if(s.connectionAttempted()){
								// I've made an attempt to connect to it
								if(selectKeys.contains(s.getChannel().keyFor(selector))){
									
									SelectionKey k = s.getChannel().keyFor(selector);
									if (k.isConnectable()) {
										boolean result = finishConnection(k);
										
										if(result==true){
											System.out.println("connected");
											s.setConnectionEstablished(true);
											
											// once you are connected, should change interest to read
											//  don't ever change interest to write unless you have 
											//  data to be written
											k.interestOps(SelectionKey.OP_READ); //changing interest to read
										}else{
											System.out.println("connection failed");
										}
									} 
									selectKeys.remove(k);
								}
							}
						}
					}else{
						// channel established sent something to you
						if(selectKeys.contains(s.getChannel().keyFor(selector))){
							SelectionKey k = s.getChannel().keyFor(selector);
							if (k.isReadable()) {
								System.out.println("\nreadable content from other server:" + s.getServerName());
								readFromServer(k, s);
								selectKeys.remove(k);
							}
						}
					}
				}
			}
			
		} catch (IOException e) {
			e.printStackTrace();
		}
	}


	/**
	 * given the ip address return the server information
	 */
	private Server lookupServerByIP(String address) {
		for(String name:serverInfo.keySet()){
			if(!name.equals(this.myServerName)){
				Server s = serverInfo.get(name);
				
				if(s.getIp().equals(address)){
					return s;
				}
			}
		}
		System.out.println("cannot find server with address=" + address);
		return null;
	}
	
	/*-----------------------------------------*/
	/* method to convert degrees to radians    */
	/*-----------------------------------------*/
	private double deg2rad(double deg) {
		return (deg * Math.PI / 180.0);
	}
	/*-----------------------------------------*/
	/* method to convert radians to degrees    */
	/*-----------------------------------------*/
	private double rad2deg(double rad) {
		return (rad * 180.0 / Math.PI);
	}
	/*------------------------------------------------------------*/
	/* method to calculate the distance between 2 GPS coordinates */
	/*------------------------------------------------------------*/
	private double distance(double lat1, double lng1, double lat2, double lng2) {
		double theta = lng1 - lng2;
		double dist = Math.sin(deg2rad(lat1)) * Math.sin(deg2rad(lat2)) + Math.cos(deg2rad(lat1)) * Math.cos(deg2rad(lat2)) * Math.cos(deg2rad(theta));
		dist = Math.acos(dist);
		dist = rad2deg(dist);
		dist = dist * 60 * 1.1515;
		dist = dist * 1.609344; 					// to calculate distance in kilo meters
		return (dist);
	}
	
	/**
	 * main
	 * @param args 
	 * args[0] = configuration file name
	 * args[1] = server name
	 */
	public static void main(String[] args) {
		if(args.length < 2){
			System.err.println("please provide two parameters");
			System.exit(0);
		}
		
		try {
			// start the selecting thread
			new Thread(new NioServer(args[0], args[1])).start();
		} catch (IOException e) {
			e.printStackTrace();
		}
	}
}
